package com.atguigu.edu.realtime.app.dwd.db;


import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 *下单事实表
 */
public class DwdTradeOrderDetail {
    public static void main(String[] args) {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(5);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //1.4 设置状态的TTL(传输的延迟 + 业务上的滞后关系)
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));
        //TODO 2.检查点相关的设置(略)
        //TODO 3.从kafka的topic_db主题中读取数据 创建动态表
        String groupId="dwd_trade_order_detail";
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_trade_order_detail"));
        //TODO 4.从kafka读取日志数据dwd_traffic_page_log(页面日志数据,为了获取数据源source_id)
        tableEnv.executeSql("create table page_log(\n" +
                "common map<String,String>,\n" +
                "page map<String,String>,\n" +
                "ts string\n" +
                ")"+MyKafkaUtil.getKafkaDDL("dwd_traffic_page_log",groupId));
        //TODO 5.过滤订单明细数据
        Table orderDetail = tableEnv.sqlQuery("\n" +
                "select\n" +
                "`data`['id'] id,\n" +
                "`data`['course_id'] course_id,\n" +
                "`data`['course_name'] course_name,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['origin_amount'] origin_amount,\n" +
                "`data`['coupon_reduce'] coupon_reduce,\n" +
                "`data`['final_amount'] final_amount,\n" +
                "`data`['create_time'] create_time,\n" +
                " date_format(data['create_time'],'yyyy-MM-dd') create_date,\n" +
                " ts\n" +
                "from topic_db \n" +
                "where `table`='order_detail'\n" +
                "and type='insert'");
        tableEnv.createTemporaryView("order_detail",orderDetail);
        //TODO 6.过滤订单数据
        Table orderInfo = tableEnv.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['order_status'] order_status,\n" +
                "`data`['out_trade_no'] out_trade_no,\n" +
                "`data`['trade_body'] trade_body,\n" +
                "`data`['session_id'] session_id,\n" +
                "`data`['province_id'] province_id \n" +
                "from topic_db \n" +
                "where `table`='order_info' \n" +
                "and type='insert'");
        tableEnv.createTemporaryView("order_info",orderInfo);

        //TODO 7 获取下单日志
        Table orderLog = tableEnv.sqlQuery("select \n" +
                "    common['sid'] session_id,\n" +
                "    common['sc'] source_id\n" +
                "from page_log\n" +
                "where page['page_id']='order'");
        tableEnv.createTemporaryView("order_log",orderLog);
        //TODO 8 关联3张表格
        Table resultTable = tableEnv.sqlQuery("select\n" +
                "od.id,\n" +
                "od.course_id,\n" +
                "od.course_name,\n" +
                "od.order_id,\n" +
                "od.user_id,\n" +
                "od.origin_amount,\n" +
                "od.coupon_reduce,\n" +
                "od.final_amount,\n" +
                "od.create_time,\n" +
                "od.create_date,\n" +
                "oi.order_status,\n" +
                "oi.out_trade_no,\n" +
                "oi.trade_body,\n" +
                "oi.session_id,\n" +
                "oi.province_id,\n" +
                "ol.source_id,\n" +
                "od.ts\n" +
                "from order_detail od\n" +
                "join order_info oi\n" +
                "on od.order_id=oi.id\n" +
                "left join order_log ol\n" +
                "on oi.session_id=ol.session_id");
        tableEnv.createTemporaryView("result_table",resultTable);

        //TODO 9.将关联的结果写到kafka主题中
        //9.1 创建动态表和要写入的主题进行映射
        tableEnv.executeSql("create table dwd_trade_order_detail(\n" +
                "id String,\n" +
                "course_id String,\n" +
                "course_name String,\n" +
                "order_id String,\n" +
                "user_id String,\n" +
                "origin_amount String,\n" +
                "coupon_reduce String,\n" +
                "final_amount String,\n" +
                "create_time String,\n" +
                "create_date String,\n" +
                "order_status String,\n" +
                "out_trade_no String,\n" +
                "trade_body String,\n" +
                "session_id String,\n" +
                "province_id String,\n" +
                "source_id String,\n" +
                "ts String,\n" +
                "primary key(id) not enforced\n" +
                ")"+MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_order_detail"));
        //9.2写入
        tableEnv.executeSql("insert into dwd_trade_order_detail select * from result_table");

    }
}
